package com.cike.sparkstudy.sql.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;

public class RDD2DataFrameProgrammatically {
    public static void main(String[] args){
        SparkConf conf = new SparkConf()
                .setMaster("local")
                .setAppName("RDD2DataFrameProgrammatically");

        JavaSparkContext sc = new JavaSparkContext(conf);

        SQLContext sqlContext = new SQLContext(sc);

        //1、构建普通RDD
        JavaRDD<String> lines = sc.textFile("/developerCodes/test/students.txt");

        JavaRDD<Row> studentRDD = lines.map(new Function<String, Row>() {
            @Override
            public Row call(String line) throws Exception {

                String[] lineSplit = line.split(",");
                return RowFactory.create(
                        Integer.valueOf(lineSplit[0]),
                        lineSplit[1],
                        Integer.valueOf(lineSplit[2]));
            }
        });


        //第二步动态构建元数据
        //  主要用于一开始不确定数据类型或者从配置文件、db中的数据类型的元数据构造
        ArrayList<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("id",DataTypes.IntegerType,true));
        structFields.add(DataTypes.createStructField("name",DataTypes.StringType,true));
        structFields.add(DataTypes.createStructField("age",DataTypes.IntegerType,true));

        StructType structType = DataTypes.createStructType(structFields);
        //第三步，使用动态构建的元数据，将RDD和DataFrame转换
        DataFrame studentDF = sqlContext.createDataFrame(studentRDD, structType);
        //后面使用DataFrame可以正常使用
        //注册
        studentDF.registerTempTable("students");

        //查询语句
        DataFrame teenagerDF = sqlContext.sql("select * from students where age<=17");

        List<Row> rows = teenagerDF.javaRDD().collect();

        for(Row row:rows){
            System.out.println(row);
        }
    }
}
